package com.hzgj.bcl.soa.rpc.server;

import com.hzgj.bcl.soa.Event;
import com.hzgj.bcl.soa.Interceptor;
import com.hzgj.bcl.soa.cicada.pool.Channels;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.soa.util.GlobalInvoke;
import com.hzgj.bcl.soa.util.NamedThreadFactory;
import com.hzgj.bcl.soa.util.ServiceLoaderUtil;
import com.hzgj.bcl.util.jmx.Management;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.Getter;

/**
 * Created by dell on 2015/3/30.
 */
public class NettyServer extends Interceptor {

    @Getter
    private final ServerConfig serverConfig;
    private final ExecutorService threadPool;
    private final EventLoopGroup bossGroup;
    private final NioEventLoopGroup workerGroup;
    private final ChannelGroup allChannels;
    private Logger logger = LoggerFactory.getLogger(getClass());
    private AtomicBoolean start = new AtomicBoolean(false);
    private ServerBootstrap bootstrap = null;

    public NettyServer(final ServerConfig serverConfig) {
        this.serverConfig = serverConfig;
        ThreadFactory serverBossTF = new NamedThreadFactory("nettyServer-boss-");
        ThreadFactory serverWorkerTF = new NamedThreadFactory("nettyServer-worker-");
        ThreadFactory tf = new NamedThreadFactory("businessThreadPool-");

        threadPool = new ThreadPoolExecutor(serverConfig.getBizThreadPoolConfig().getMin(),
                serverConfig.getBizThreadPoolConfig().getMax(),
                serverConfig.getBizThreadPoolConfig().getKeepLiveInSecond(),
                TimeUnit.SECONDS, new LinkedBlockingQueue<>(serverConfig.getBizThreadPoolConfig().getQueueSize()), tf);

        bossGroup = new NioEventLoopGroup(serverConfig.getBossGroupThreads(), serverBossTF);
        workerGroup = new NioEventLoopGroup(serverConfig.getWorkerGroupThreads(), serverWorkerTF);
        workerGroup.setIoRatio(serverConfig.getIoRatio());
        allChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        ConnectionHandler connectionHandler = new ConnectionHandler(allChannels);
        ServerStat.getInstance().addConnectionHandler(serverConfig.getPort(), connectionHandler);
        Management.register(connectionHandler, Constants.getConnectionHandlerObjectName(serverConfig.getPort()));
        bootstrap = new ServerBootstrap().group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.SO_REUSEADDR, serverConfig.isReuseAddress())
                .option(ChannelOption.TCP_NODELAY, serverConfig.isNodelay())
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        IdleStateHandler idleStateHandler = new IdleStateHandler(serverConfig.getKeepAliveTime(), serverConfig.getKeepAliveTime(), serverConfig.getKeepAliveTime());
                        pipeline.addLast("idleState", idleStateHandler);
                        pipeline.addLast("idleProcess", new IdleHandler());
                        pipeline.addLast("connectionHandler", connectionHandler);

                        pipeline.addLast("decoder", ServiceLoaderUtil.getDecoder(serverConfig.getProtocol()));
                        pipeline.addLast("encoder", ServiceLoaderUtil.getEncoder(serverConfig.getProtocol()));
                        pipeline.addLast("handler", new NettyServerHandler(NettyServer.this, threadPool));
                    }
                });
    }

    @Override
    public Object down(Event evt) {
        Message message = (Message) evt.getArg();
        Long req_id = (Long) message.getHeader(Constants.MESSAGE_SID);
        Channel channel = Channels.getChannel(req_id);
        if (channel != null) {
            channel.writeAndFlush(message);
        } else {
            throw new RuntimeException("can't find response channel by requestId = " + req_id);
        }
        return null;
    }

    @Override
    public void start() throws Exception {
        if (!start.compareAndSet(false, true)) {
            return;
        }
        logger.info(serverConfig.getName() + " server started,listen at: " + serverConfig.getPort());
        InetSocketAddress inetSocketAddress = new InetSocketAddress(serverConfig.getPort());
        ChannelFuture channelFuture = bootstrap.bind(inetSocketAddress).sync().addListener(future -> {
            if (future.isSuccess()) {
                GlobalInvoke.invoke(new GlobalInvoke.ServerStartedInvokeEvent(serverConfig.getName()));
            } else {
                logger.error("{}", future.cause());
            }
        });
        if (serverConfig.isSuspend()) {
            channelFuture.channel().closeFuture().sync();
        } else {
            channelFuture.sync();
        }
    }

    @Override
    public void stop() {
        start.set(false);
        threadPool.shutdown();
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        allChannels.close().awaitUninterruptibly();
        logger.info("server stopped");
    }
}
